package com.atguigu.actual.edu0417.app.dwd.db;

import com.atguigu.actual.edu0417.app.SqlBaseApp;
import com.atguigu.actual.edu0417.unils.MyKafkaUtil;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

/**
 * @author: techyang
 * @since: 2023-10-20 18:06
 * @description: 交易域支付成功事务事实表
 **/
public class DwdTradePayDetailSuc extends SqlBaseApp {

    public static void main(String[] args) {
        DwdTradePayDetailSuc dwdTradePayDetailSuc = new DwdTradePayDetailSuc();
        dwdTradePayDetailSuc.base(args);
    }

    @Override
    public void process(StreamTableEnvironment tableEnv) {

        // 设置状态有效时间为 15分钟 + 10秒钟
        tableEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(15 * 60 + 10));

        // 从topic_db读取数据
        tableEnv.executeSql(" create table topic_db (" +
                " `database` string,\n" +
                "`table` string,\n" +
                "`type` string,\n" +
                "`ts` string,\n" +
                "`data` map<string,string>,\n" +
                "`old` map<string,string>,\n" +
                "proc_time as PROCTIME(), " +
                "event_time as TO_TIMESTAMP(FROM_UNIXTIME(cast(ts as bigint))), " +
                "watermark for event_time as event_time " +
                ") " + MyKafkaUtil.getKafkaDDL("topic_db","dwd_trade_pay_detail_suc_group"));

        // 过滤出支付成功的数据
        Table paySucTable = tableEnv.sqlQuery(" select " +
                "data['id'] id, " +
                "data['out_trade_no'] out_trade_no, " +
                "data['order_id'] order_id, " +
                "data['alipay_trade_no'] alipay_trade_no, " +
                "data['payment_type'] payment_type, " +
                "data['payment_status'] payment_status, " +
                "data['callback_time'] callback_time, " +
                "proc_time, " +
                "event_time " +
                "from topic_db " +
                "where `table` = 'payment_info' and `type` = 'insert' " +
                " and data['payment_status'] = '1602' ");
        tableEnv.createTemporaryView("pay_suc",paySucTable);
        // tableEnv.executeSql("select *  from pay_suc ").print();

        // 读取 dwd_trade_order_detail 下单数据
        tableEnv.executeSql(" create table dwd_trade_order_detail (" +
                "id string, " +
                "course_id string, " +
                "course_name string, " +
                "order_id string, " +
                "user_id string, " +
                "origin_amount string, " +
                "coupon_reduce string, " +
                "final_amount string, " +
                "order_status string, " +
                "out_trade_no string, " +
                "session_id string, " +
                "province_id string, " +
                "ts string, " +
                "event_time as TO_TIMESTAMP(FROM_UNIXTIME(cast(ts as bigint))), " +
                "watermark for event_time as event_time " +
                ") " + MyKafkaUtil.getKafkaDDL("dwd_trade_order_detail","dwd_trade_pay_detail_suc_group"));

        // tableEnv.executeSql(" select * from dwd_trade_order_detail ").print();

        //关联支付成功数据与下单数据
        Table joinTable = tableEnv.sqlQuery(" select " +
                "od.id, " +
                "od.course_id, " +
                "od.course_name, " +
                "od.order_id, " +
                "od.user_id, " +
                "od.origin_amount, " +
                "od.coupon_reduce, " +
                "od.final_amount, " +
                "od.order_status, " +
                "od.out_trade_no, " +
                "od.session_id, " +
                "od.province_id, " +
                "od.ts, " +
                "ps.payment_type, " +
                "ps.payment_status, " +
                "ps.callback_time " +
                "from dwd_trade_order_detail od,pay_suc ps " +
                "where od.out_trade_no = ps.out_trade_no " +
                "and od.event_time >= ps.event_time - INTERVAL '15' MINUTE " +
                "and od.event_time <= ps.event_time + INTERVAL '5' SECOND");
        tableEnv.createTemporaryView("join_table",joinTable);
        tableEnv.executeSql(" select * from join_table ");

        // 创建动态表
        tableEnv.executeSql(" create table dwd_trade_pay_detail_suc (" +
                "id string, " +
                "course_id string, " +
                "course_name string, " +
                "order_id string, " +
                "user_id string, " +
                "origin_amount string, " +
                "coupon_reduce string, " +
                "final_amount string, " +
                "order_status string, " +
                "out_trade_no string, " +
                "session_id string, " +
                "province_id string, " +
                "ts string, " +
                "payment_type string, " +
                "payment_status string, " +
                "callback_time string, " +
                "primary key(id) not enforced" +
                ") " + MyKafkaUtil.getUpsertKafkaDDL("dwd_trade_pay_detail_suc"));

        // 写入 dwd_trade_pay_detail_suc 主题
        tableEnv.executeSql(" insert into dwd_trade_pay_detail_suc select * from join_table ");

    }
}
